package tuijianhaoyou;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import java.io.File;
import java.io.IOException;
import java.util.*;


public class Fridents {
    public static class FridentsMap extends Mapper<LongWritable, Text, Text, Text> {
        Text v = new Text();
        Text z = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//            String line = value.toString();
//            String[] carry  = line.split(":");
            String[] carry = value.toString().split(":");
            String car = carry[0];
            String[] carr = carry[1].split(",");
            v.set(car);
            for (String t : carr) {
                z.set(t);
                context.write(v, z);
            }
        }
    }

    public static class FridentsReduce extends Reducer<Text, Text, Text, Text> {
        @Override
        protected void reduce(Text friend, Iterable<Text> values, Context context) throws IOException, InterruptedException {
            ArrayList<String> strings = new ArrayList<String>();
            //把值放到strings中
            for (Text aa : values) {
                strings.add(aa.toString());
            }
            Collections.sort(strings);
            for (int i = 0; i < strings.size() - 1; i++) {
                for (int j = i + 1; j < strings.size(); j++) {
                    context.write(new Text(strings.get(i) + "-" + strings.get(j)), friend);
                }
            }

        }
    }

    public static void main(String[] args) throws ClassNotFoundException, InterruptedException {
        Configuration conf = new Configuration();
        try {
            Job job = Job.getInstance(conf);
            job.setJarByClass(Fridents.class);

            job.setMapperClass(FridentsMap.class);
            job.setReducerClass(FridentsReduce.class);

            job.setMapOutputValueClass(Text.class);
            job.setMapOutputValueClass(Text.class);

            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            File file = new File("f:/all/a/tuijianhaoyou/xx");
            if (file.exists()) {
                FileUtil.fullyDelete(file);
            }

            FileInputFormat.setInputPaths(job, new Path("f:/all/a/tuijianhaoyou/xx.txt"));
            FileOutputFormat.setOutputPath(job, new Path("f:/all/a/tuijianhaoyou/xx"));
            job.setNumReduceTasks(1);

            System.exit(job.waitForCompletion(true) ? 0 : 1);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
